package com.sjc.lesson02.state.keyedState.function;

import com.google.common.collect.Lists;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Collections;

/**
 * ListState<T> ：这个状态为每一个 key 保存集合的值
 *      get() 获取状态值
 *      add() / addAll() 更新状态值，将数据放到状态中
 *      clear() 清除状态
 */
public class CountWindowAverageWithListState extends RichFlatMapFunction<Tuple2<Long,Long>, Tuple2<Long,Double>> {
    // managed keyed state
    //1. ListState 保存的是对应的一个 key 的出现的所有的元素
    private ListState<Tuple2<Long,Long>> elementStateByKey;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 注册状态
        ListStateDescriptor<Tuple2<Long, Long>> descriptor = new ListStateDescriptor<Tuple2<Long, Long>>(
                        "average", // 状态的名字
                        Types.TUPLE(Types.LONG, Types.LONG)); // 状态存储的数据类型
        elementStateByKey = getRuntimeContext().getListState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> element, Collector<Tuple2<Long, Double>> collector) throws Exception {
        // 拿到当前key的值
       Iterable<Tuple2<Long, Long>> currentState = elementStateByKey.get();
        // 如果状态值还没有初始化，则初始化
        if(currentState == null){
            elementStateByKey.addAll(Collections.emptyList());
        }
        // 更新状态
        elementStateByKey.add(element);

        // 判断，如果当前的key出现了3次，则需要计算平均值，并且输出
        ArrayList<Tuple2<Long, Long>> allElements = Lists.newArrayList(elementStateByKey.get());
        if(allElements.size() >= 3){
            long count =0; long sum = 0;
            for(Tuple2<Long,Long> ele : allElements){
                count++;
                sum += ele.f1;
            }

            double avg = sum/count;
            collector.collect(Tuple2.of(element.f0, avg));

            // 清除状态
            elementStateByKey.clear();
        }
    }
}
